package com.sfx.concurrent.demo.BlockingQueue.ThreadPoll;

import lombok.extern.slf4j.Slf4j;

import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * 任务生产者生产的任务对象交给线程池
 * 如果没有线程就创建线程执行
 * 如果有线程但线程数量不足,就像任务放到阻塞队列中
 */

/**
 * 拒绝策略->定义一个方法,让调用者决定怎么执行->队列+任务
 * 函数式接口
 */
@FunctionalInterface
interface RejectPolicy<T> {
    void reject(MyBlockingQueue<T> queue, T task);
}

@Slf4j(topic = "c.MyThreadPool")
public class MyThreadPool {
    // 阻塞队列,用于存放,执行不了的任务
    private MyBlockingQueue<Runnable> taskQueue;
    // 核心线程数
    private int corePoolSize;
    // 等待任务执行的超时时间
    private long timeUnit;
    // 时间单位
    private TimeUnit unit;
    // 拒绝策略
    private RejectPolicy rejectPolicy;
    // 维护线程池中的线程->一个线程集合
    private HashSet<Thread> workers = new HashSet<>();

    public MyThreadPool(int corePoolSize, long timeUnit, TimeUnit unit, RejectPolicy rejectPolicy,int capacity) {
        this.corePoolSize = corePoolSize;
        this.timeUnit = timeUnit;
        this.unit = unit;
        this.rejectPolicy = rejectPolicy;
        taskQueue = new MyBlockingQueue<>(capacity);
    }

    /**
     * 提交任务并执行
     * @param task
     */
    public void execute(Runnable task) {
        synchronized (workers) {
            // 线程数量 < 核心线程数
            if(workers.size() < this.corePoolSize) {
                // 创建一个Workers,来执行任务
                Worker worker = new Worker(task);
                worker.start();
                log.debug("新增worker{},任务{}",worker,task);
                workers.add(worker);
            } else {
                // 尝试放入阻塞队列,放不进去执行拒绝策略
                taskQueue.tryPut(rejectPolicy,task);
            }
        }
    }

    //工作线程
    class Worker extends Thread {
        Runnable task;
        public Worker(Runnable task) {
            this.task = task;
        }

        /**
         * 执行工作线程中的任务
         */
        @Override
        public void run() {
            // 不断地执行任务,任务执行完之后,去阻塞队列中取出任务
            while(task != null || taskQueue.take() != null) {

            }
        }
    }

}